package com.yixing.tech.wms.statistics.stream;

import com.yixing.tech.wms.statistics.domain.entity.InventoryInitialization;
import com.yixing.tech.wms.stock.out.StockOutConstants;
import com.yixing.tech.wms.stock.out.domain.Inventory;
import com.yixing.tech.wms.stock.out.domain.StockOutOperation;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest;
import org.springframework.data.mongodb.core.messaging.MessageListenerContainer;
import org.springframework.data.mongodb.core.query.Criteria;

import javax.annotation.Resource;

@Configuration
public class ChangeStreamInitializer implements CommandLineRunner {

    @Resource
    InventoryInfoListener inventoryInfoListener;

    @Resource
    InventoryInitializationListener inventoryInitializationListener;

    @Resource
    StockOutOperationListener stockOutOperationListener;

    @Resource
    private MessageListenerContainer messageListenerContainer;

    @Override
    public void run(String... args) throws Exception {
        ChangeStreamRequest<Inventory> request1 = ChangeStreamRequest
                .builder(inventoryInfoListener)
                .collection(StockOutConstants.INVENTORY_TEMPLATE)
                .filter(Aggregation.newAggregation(Aggregation.match(Criteria.where("operationType").in("insert", "update", "replace", "delete"))))
                .build();

        ChangeStreamRequest<InventoryInitialization> request2 = ChangeStreamRequest
                .builder(inventoryInitializationListener)
                .collection(StockOutConstants.INVENTORY_INITIALIZATION_TEMPLATE)
                .filter(Aggregation.newAggregation(Aggregation.match(Criteria.where("operationType").in("insert", "update", "replace", "delete"))))
                .build();

        ChangeStreamRequest<StockOutOperation> request3 = ChangeStreamRequest
                .builder(stockOutOperationListener)
                .collection(StockOutConstants.STOCK_OUT_OPERATION_TEMPLATE)
                .filter(Aggregation.newAggregation(Aggregation.match(Criteria.where("operationType").in("insert", "update", "replace", "delete"))))
                .build();

        messageListenerContainer.register(request1, Inventory.class);
        messageListenerContainer.register(request2, InventoryInitialization.class);
        messageListenerContainer.register(request3, StockOutOperation.class);
    }
}
